package com.lobstar.xrpc.server.threads;

import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lobstar.xrpc.common.po.SMessage;
import com.lobstar.xrpc.server.listener.DefaultServiceListener;
import com.lobstar.xrpc.server.so.ServerWork;
import com.lobstar.xrpc.service.base.ServiceInvoker;
import com.lobstar.xrpc.service.utils.JsonUtils;

import io.netty.channel.ChannelHandlerContext;

public class ServerWorkThread extends Thread {

	private Logger logger = LoggerFactory.getLogger(ServerWorkThread.class);
	private BlockingQueue<ServerWork> workQueue;
	private int threadSize = 1;
	private ExecutorService runner;
	private ExecutorService holder;
	private ServiceInvoker invoker;

	private ServerWorkThread(ServiceInvoker invoker) {
		workQueue = new ArrayBlockingQueue<>(500);
	}
	private ServerWorkThread(int size,ServiceInvoker invoker) {
		this(invoker);
		this.threadSize = size;
	}

	public static ServerWorkThread build(ServiceInvoker invoker) {
		ServerWorkThread swt = new ServerWorkThread(invoker);
		return swt;
	}

	public static ServerWorkThread build(int size,ServiceInvoker invoker) {
		ServerWorkThread swt = new ServerWorkThread(size,invoker);
		return swt;
	}

	public void addWork(ChannelHandlerContext ctx, SMessage msg) {
		workQueue.add(ServerWork.build(msg, ctx));
	}
	public void start() {
		runner = Executors.newFixedThreadPool(threadSize);
		holder = Executors.newSingleThreadExecutor();
		holder.execute(this);
	}

	@Override
	public void run() {
		while(true) {
			try {
				ServerWork serverWork = workQueue.take();
				runner.execute(()->{
					handleRequest(serverWork.getCtx(),serverWork.getsMessage());
				});
			} catch (InterruptedException e) {
				logger.error("server work exception",e);
			}
		}
	}

	public int getThreadSize() {
		return threadSize;
	}

	public void setThreadSize(int threadSize) {
		this.threadSize = threadSize;
	}

	private void handleRequest(ChannelHandlerContext ctx, SMessage msg) {
		Object ret = null;
		Object ex = null;
		//记录msg的id标识返回对象
		try {
			if(invoker.hasMethod(msg)) {
				try {
					ret = invoker.invoke(msg);
					msg.addAttribute("_ret", JsonUtils.toJson(ret));
				}catch(Exception e) {
					if(e.getClass() == InvocationTargetException.class) {
						ex = ((InvocationTargetException)e).getTargetException().toString();
					}else {
						ex = e.toString();
					}
				}
			}else {
				ex = "no method";
			}
			if(ex != null) {
				msg.addAttribute("_exception", ex);
			}
			ctx.channel().writeAndFlush(msg.buildResponse());
		}catch(Exception e) {
			logger.error("request",e);
		}
	}
}
